package com.ymt.reliable.message.confirm.service.impl;

import com.ymt.common.tools.DateUtils;
import com.ymt.producer.service.BusinessProducerConfirmService;
import com.ymt.reliable.message.confirm.config.MessageConfirmConfig;
import com.ymt.reliable.message.confirm.service.MessageConfirmService;
import com.ymt.reliable.message.confirm.task.Task;
import com.ymt.reliable.message.domain.TransactionMessage;
import com.ymt.reliable.message.service.TransactionMessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;

@Service
public class MessageConfirmServiceImpl implements MessageConfirmService, Runnable {
    private static final Logger LOGGER = LoggerFactory.getLogger(MessageConfirmServiceImpl.class);
    private static final String THREDNAME = "确认消息重试主线程";
    private static DelayQueue<Task> QUEUE = new DelayQueue<>();

    @Autowired
    private TransactionMessageService transactionMessageService;
    @Autowired
    private BusinessProducerConfirmService businessProducerConfirmService;
    @Autowired
    private MessageConfirmConfig config;

    @Override
    public void addTask(Task task) {
        synchronized (task) {
            if (!QUEUE.contains(task)) {
                SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                LOGGER.info(">>> 任务加入本地消息确认延迟队列!taskId:{},createTime:{},submitTime:{},剩余时间:{}",
                        task.getTaskId(),
                        DateUtils.formatDate(task.getMessage().getCreateTime()),
                        DateUtils.formatDate(task.getSubmitTime()),
                        task.getDelay(TimeUnit.MILLISECONDS));
                QUEUE.add(task);
            }
        }
    }

    @Override
    public Integer queueSize() {
        return QUEUE.size();
    }

    @Override
    public void run() {
        while (true) {
            Task task = null;
            try {
                task = QUEUE.take();
                LOGGER.info(">>> 任务到期取出任务,任务编号:{},当前时间:{}", task.getTaskId(), DateUtils.currentDateTimeToString());
            } catch (Exception e) {
                e.printStackTrace();
            }
            if (task != null) {
                TransactionMessage message = task.getMessage();
                try {
                    //确认消息
                    if (businessProducerConfirmService.confirm(message.getField1())) {
                        //确认次数判断
                        TransactionMessage copy = new TransactionMessage();
                        copy.setId(message.getId());
                        copy.setMessageConfirmTimes(message.getMessageConfirmTimes());
                        copy.addMessageConfirmTimes();
                        if (copy.getMessageConfirmTimes() >= config.getMaxconfirmtimes()) {
                            continue;
                        }

                        transactionMessageService.update(copy);
                        transactionMessageService.confirmAndSend(message.getMessageId());
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @PostConstruct
    public void start() {
        Thread thread = new Thread(this);
        thread.setName(THREDNAME);
        thread.start();
        LOGGER.info(">>> 消息确认重试主线程启动成功......");
    }
}
